/*
 *
 *  * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 *
 *
 */

package com.iwindplus.log.server.listener;

import cn.hutool.core.text.CharSequenceUtil;
import com.fasterxml.jackson.core.type.TypeReference;
import com.iwindplus.base.domain.dto.MessageBaseDTO;
import com.iwindplus.base.util.JacksonUtil;
import com.iwindplus.log.domain.dto.GatewayLogDTO;
import com.iwindplus.log.server.service.GatewayLogService;
import jakarta.annotation.Resource;
import java.util.List;
import java.util.Objects;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 网关日志监听器.
 *
 * @author zengdegui
 * @since 2025/08/24 13:04
 */
@Slf4j
@Component
public class GatewayLogListener {

    @Resource
    private GatewayLogService gatewayLogService;

    @KafkaListener(
        topics = {"${spring.kafka.configs[0].topic-name}"},
        groupId = "${spring.kafka.configs[0].group-id}"
    )
    public void listenSingle(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        records.stream().filter(m -> Objects.nonNull(m) && CharSequenceUtil.isNotBlank(m.value()))
            .forEach(record -> processMessage(record));

        ack.acknowledge();
    }

    private void processMessage(ConsumerRecord<String, String> record) {
        try {
            final String value = record.value();
            log.info("网关日志消息消费参数={}", value);
            final MessageBaseDTO<GatewayLogDTO> messageDTO = JacksonUtil.parseObject(value, new TypeReference<>() {
            });
            this.gatewayLogService.saveBatch(List.of(messageDTO.getData()));
        } catch (Exception ex) {
            log.error("网关日志消息消费失败={}", ex);
        }
    }
}
